package com.mini.cn.goods.io.netty.ws;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.codec.http.websocketx.*;
import org.apache.commons.lang3.ArrayUtils;

import java.util.List;

/**
 * Created with IntelliJ IDEA.

 * User: fly
 * Date: 2020-01-13
 * Time: 11:07 上午
 */
@ChannelHandler.Sharable
final class WebSocketServerHandler extends SimpleChannelInboundHandler<WebSocketFrame> {

    final IWsHandel wsHandel;

    final boolean isLastHandler;

    public WebSocketServerHandler(IWsHandel wsHandel, boolean isLastHandler) {
        this.wsHandel = wsHandel;
        this.isLastHandler = isLastHandler;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame msg) throws Exception {
        handleWebSocketFrame(ctx, msg);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        if (wsHandel != null) {
            WebSocketSession session = ctx.channel().attr(WsConfig.SESSION_KEY).get();
            wsHandel.onError(session, cause);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if (wsHandel != null) {
            WebSocketSession session = ctx.channel().attr(WsConfig.SESSION_KEY).get();
            wsHandel.onClose(session);
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (wsHandel != null) {
            WebSocketSession session = ctx.channel().attr(WsConfig.SESSION_KEY).get();
            wsHandel.onEvent(session, evt);
        }
    }

    private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
        if (frame instanceof TextWebSocketFrame) {
            if (wsHandel != null) {
                WebSocketSession session = ctx.channel().attr(WsConfig.SESSION_KEY).get();
                wsHandel.onMessage(session, ((TextWebSocketFrame) frame).text());
            }
            return;
        }
        if (frame instanceof PingWebSocketFrame) {
            ctx.writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
            if (wsHandel != null) {
                WebSocketSession session = ctx.channel().attr(WsConfig.SESSION_KEY).get();
                wsHandel.onPingMessage(session);
            }
            return;
        }
        if (frame instanceof CloseWebSocketFrame) {
            ctx.writeAndFlush(frame.retainedDuplicate()).addListener(ChannelFutureListener.CLOSE);
            return;
        }

        if (frame instanceof BinaryWebSocketFrame) {
            ByteBuf content = frame.content();
            if (isLastHandler) {
                if (wsHandel != null) {
                    WebSocketSession session = ctx.channel().attr(WsConfig.SESSION_KEY).get();
                    byte[] bytes = new byte[content.readableBytes()];
                    content.readBytes(bytes);
                    wsHandel.onBinary(session, bytes);
                }
            } else {
                content.retain();
                ctx.fireChannelRead(content);
            }
            return;
        }
        //解决continution的问题
        if (frame instanceof ContinuationWebSocketFrame) {
            if (isLastHandler) {
                if (wsHandel != null) {
                    handleContinuation(ctx, frame);
                }
            } else {
                ByteBuf content = frame.content();
                WebSocketSession session = ctx.channel().attr(WsConfig.SESSION_KEY).get();
                ByteBuf byteBuf = session.getByteBuf();
                if (frame.isFinalFragment()) {
                    byteBuf.writeBytes(content);
                    ctx.fireChannelRead(byteBuf);
                    session.setByteBuf(null);
                } else {
                    if (byteBuf == null) {
                        //第一次
                        byteBuf = Unpooled.directBuffer();
                    }
                    byteBuf.writeBytes(content);
                }
            }
            return;
        }
        if (frame instanceof PongWebSocketFrame) {
            return;
        }
    }

    /**
     * 校验整体包的大小
     *
     * @param bytes
     * @return
     */
    private boolean checkMaxLengh(List<byte[]> bytes, int increase) {
        int totalSize = 0;
        for (byte[] bs : bytes) {
            totalSize += bs.length;
        }
        totalSize += increase;
        return totalSize < WsConfig.MAXSIZE;
    }

    /**
     * 发送错误信息
     *
     * @param session
     */
    private void sendErrorRespone(WebSocketSession session) {
        ChannelFuture future = session.sendText("package too large >" + WsConfig.MAXSIZE);
        future.addListener(ChannelFutureListener.CLOSE);
    }

    private void handleContinuation(ChannelHandlerContext ctx, WebSocketFrame frame) {
        WebSocketSession session = ctx.channel().attr(WsConfig.SESSION_KEY).get();
        ByteBuf content = frame.content();
        //当前包大小
        int bnumber = content.readableBytes();
        List<byte[]> list = session.getBytes();
        //合包进行校验
        if (!checkMaxLengh(list, bnumber)) {
            list.clear();
            sendErrorRespone(session);
            return;
        }
        byte[] bytes = new byte[bnumber];
        //读到bytes数组
        content.readBytes(bytes);
        list.add(bytes);
        if (frame.isFinalFragment()) {
            if (wsHandel != null) {
                //否则整合包以后callback回调
                byte[] allBytes = new byte[0];
                for (byte[] bs : list) {
                    allBytes = ArrayUtils.addAll(allBytes, bs);
                }
                wsHandel.onBinary(session, allBytes);
            }
            list.clear();
        }
    }
}
